package com.chukun.flink.stream.window.join.watermarker;

import com.chukun.flink.stream.window.join.bean.ClickBean;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

import java.sql.Timestamp;

/**
 * @author chukun
 * @version 1.0.0
 * @description watermarker 生成器
 *
 *   数据之间存在最大固定延迟的时间戳分配器 : {@link WatermarkStrategy#forBoundedOutOfOrderness(java.time.Duration);}
 *
 *   单调递增时间戳分配器 :  {@link WatermarkStrategy#forMonotonousTimestamps();}
 *
 *
 * @createTime 2022年05月26日 23:45:00
 */
public class ClickBeanWatermarkerGenerator implements WatermarkGenerator<ClickBean> {

    /**
     * 每来一条事件数据调用一次，可以检查或者记录事件的时间戳，或者也可以基于事件数据本身去生成 watermark。
     * @param clickBean
     * @param l
     * @param watermarkOutput
     */

    private Timestamp vistTime;

    @Override
    public void onEvent(ClickBean clickBean, long timestamp, WatermarkOutput watermarkOutput) {
        vistTime = clickBean.getVisitTime();
    }

    /**
     * 周期性的调用，也许会生成新的 watermark，也许不会。
     *
     * <p>调用此方法生成 watermark 的间隔时间由 {@link ExecutionConfig#getAutoWatermarkInterval()} 决定。
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
        watermarkOutput.emitWatermark(new Watermark(vistTime.getTime() - 1));
    }
}
